package day08;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import po.UserEvent;

import java.util.HashMap;

/**
 * @Description: 使用状态机 进行连续三次登录失败的判断
 * @Author: ZYX
 * @Date: 2022/2/18 16:17
 * @Version: 1.0
 */
public class Demo05 {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<UserEvent> stream = env.fromElements(
                new UserEvent("user-1", "fail", 1000L),
                new UserEvent("user-1", "fail", 2000L),
                new UserEvent("user-1", "fail", 3000L),
                new UserEvent("user-2", "success", 3000L),
                new UserEvent("user-1", "fail", 4000L))
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<UserEvent>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<UserEvent>() {
                                    @Override
                                    public long extractTimestamp(UserEvent userEvent, long l) {
                                        return userEvent.timestamp;
                                    }
                                })
                );

        stream.keyBy(obj -> obj.userId)
                .process(new KeyedProcessFunction<String, UserEvent, String>() {
                    private HashMap<Tuple2<String,String>,String> stateMachine = new HashMap<>();
                    private ValueState<String> currentState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        stateMachine.put(Tuple2.of("INITIAL","success"),"SUCCESS");
                        stateMachine.put(Tuple2.of("INITIAL","fail"),"s1");
                        stateMachine.put(Tuple2.of("s1","success"),"SUCCESS");
                        stateMachine.put(Tuple2.of("s1","fail"),"s2");
                        stateMachine.put(Tuple2.of("s2","success"),"SUCCESS");
                        stateMachine.put(Tuple2.of("s2","fail"),"FAIL");

                        currentState = getRuntimeContext().getState(
                                new ValueStateDescriptor<String>("current-state", Types.STRING)
                        );
                    }

                    @Override
                    public void processElement(UserEvent userEvent, Context context, Collector<String> out) throws Exception {
                        if (currentState.value() == null) {
                            currentState.update("INITIAL");
                        }

                        // 计算将要跳转到的状态
                        String nextState = stateMachine.get(Tuple2.of(currentState.value(), userEvent.eventType));
                        if ("FAIL".equals(nextState)){
                            out.collect("用户："+userEvent.userId+"连续三次登录失败！");
                            currentState.update("s2");
                        }else if ("SUCCESS".equals(nextState)){
                            currentState.clear();
                        }else {
                            currentState.update(nextState);
                        }
                    }
                }).print();


        env.execute();

    }

}
